package chapter11

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DataTypes, StructType}

/**
 * author: yuhui
 * descriptions:
 * 1）通过Spark SQL来处理一个包含用户登录信息的CSV文件
 * 2）并且通过自定义用户定义函数（UDF）来计算每个登录记录的地理位置信息的GeoHash编码。
 * 3）通过GeoHash编码得到地理位置
 * date: 2024 - 11 - 28 3:45 下午
 */
object UDF02 {

  def main(args: Array[String]): Unit = {
    val session = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getSimpleName)
      .getOrCreate()

    import session.implicits._
    import org.apache.spark.sql.functions._

    val schema = new StructType()
      .add("uid", DataTypes.StringType)
      .add("login", DataTypes.StringType)
      .add("dt", DataTypes.StringType)
      .add("lat", DataTypes.DoubleType)
      .add("lng", DataTypes.DoubleType)

    val frame = session.read
      .schema(schema)
      .csv("BookData/input/11user_login.txt")

    frame.createTempView("tb_log")

    val schema2 = new StructType()
      .add("province", DataTypes.StringType)
      .add("city", DataTypes.StringType)
      .add("district", DataTypes.StringType)
      .add("lat", DataTypes.DoubleType)
      .add("lng", DataTypes.DoubleType)

    val frame2 = session.read
      .schema(schema2)
      .csv("BookData/input/11area_data.txt")

    frame2.createTempView("tb_area")

    /**
     * 自定义函数  UDF  逐行计算函数
     * 1 定义一个scala函数
     * 2 注册到spark-sql中
     */
    val f = (lat: Double, lng: Double) => {
      GeoHash.withCharacterPrecision(lat, lng, 6).toBase32
    }

    // 注册
    session.udf.register("my_geo", f)
    session.sql(
      """
        |WITH aa AS (
        |    SELECT
        |        uid,
        |        login,
        |        dt,
        |        my_geo(lat, lng) AS geo_str_log
        |    FROM
        |        tb_log
        |),
        |bb AS (
        |    SELECT
        |        province,
        |        city,
        |        district,
        |        my_geo(lat, lng) AS geo_str_area
        |    FROM
        |        tb_area
        |)
        |SELECT
        |    aa.uid,
        |    aa.login,
        |    aa.dt,
        |    bb.province,
        |    bb.city,
        |    bb.district
        |FROM
        |    aa
        |LEFT JOIN
        |    bb
        |ON
        |    aa.geo_str_log = bb.geo_str_area;
        |""".stripMargin).show()
  }

}


